package nsq

import (
	"api/config"
	"github.com/nsqio/go-nsq"
	"log"
)

var Producer *nsq.Producer
var hostProducer string
var hostSubscriber string

func init() {
	conf := config.GetCore()
	hostProducer = conf.NsqPubAddr
	hostSubscriber = conf.NsqSubAddr
	CreateProducer()
}

func CreateProducer() {
	conf := nsq.NewConfig()
	producer, err := nsq.NewProducer(hostProducer, conf)
	if err != nil {
		log.Fatal(err)
	}
	Producer = producer
}

func GetNsqProducer() *nsq.Producer {
	return Producer
}

type myMessageHandler struct {
	Handler func(content []byte) error
}

// HandleMessage implements the Handler interface.
func (h *myMessageHandler) HandleMessage(m *nsq.Message) error {
	if len(m.Body) == 0 {
		// Returning nil will automatically send a FIN command to NSQ to mark the message as processed.
		// In this case, a message with an empty body is simply ignored/discarded.
		return nil
	}

	// do whatever actual message processing is desired
	err := h.Handler(m.Body)

	// Returning a non-nil error will automatically send a REQ command to NSQ to re-queue the message.
	return err
}

func CreateNsqHandler(topic string, channel string, handler func(content []byte) error) {
	// Instantiate a consumer that will subscribe to the provided channel.
	conf := nsq.NewConfig()
	consumer, err := nsq.NewConsumer(topic, channel, conf)
	if err != nil {
		log.Fatal(err)
	}

	// Set the Handler for messages received by this Consumer. Can be called multiple times.
	// See also AddConcurrentHandlers.
	consumer.AddHandler(&myMessageHandler{
		Handler: handler,
	})

	// Use nsqlookupd to discover nsqd instances.
	// See also ConnectToNSQD, ConnectToNSQDs, ConnectToNSQLookupds.
	err = consumer.ConnectToNSQLookupd(hostSubscriber)
	if err != nil {
		log.Fatal(err)
	}

	// wait for signal to exit
	//sigChan := make(chan os.Signal, 1)
	//signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	//<-sigChan

	// Gracefully stop the consumer.
	//consumer.Stop()
}
